package com.yupi.springbootinit.rabbitmq;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * @Description
 * @Author minify
 * @Date 2023/12/17 23:39
 * @Project mybi-backend
 */
public class ConsumerMulti {

    private final static String QUEUE_NAME_MUTI = "hello_muti";

    /**
     * 消息消费者1对多
     */
    public static void main(String[] args) throws Exception {
        // 消费者消费消息
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置RabbitMQ相关信息
        factory.setHost("localhost");
        // 创建一个新的连接
        Connection connection = factory.newConnection();
        // 创建一个通道
        Channel channel = connection.createChannel();
        // 单个消费者处理任务的积压数
        channel.basicQos(1);

        for (int i = 0; i < 2; i++) {
            // 声明要关注的队列
            channel.queueDeclare(QUEUE_NAME_MUTI, true, false, false, null);
            System.out.println("Consumer Waiting Received messages");

            int finalI = i;
            DeliverCallback defaultConsumer = (consumerTag , delivery) -> {
                try {
                    TimeUnit.SECONDS.sleep(1);
                    System.out.println("Consumer"+ finalI +" Received '" + new String(delivery.getBody(), "UTF-8") + "'");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    // 手动确认消息
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                }
            };

            // DefaultConsumer类实现了Consumer接口，通过传入一个频道，
            // 如果频道中有消息，就会执行回调函数handleDelivery
            // 会阻塞在这里，等待消息
            channel.basicConsume(QUEUE_NAME_MUTI, true, defaultConsumer,(consumerTag, message) -> {});
        }
    }
}
